package dyyx;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class SocketWindowWordCount {

	 public static void main(String[] args) throws Exception{
	        String hostname = "localhost" ;
	        String delimiter = "\n" ;
	        int port = 9000;
	        // parameterTool 控制台获取参数
	        try {
	            ParameterTool parameterTool = ParameterTool.fromArgs(args);
	            port = parameterTool.getInt("port") ;
	        }catch (Exception e){     
	            port = 9000 ;
	        }
	        
	        System.out.println("hostname="+hostname+",port="+port);

	        // 1 获取运行环境
	        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	        // 2 连接socket获取输入的数据
	        DataStream<String> text = env.socketTextStream(hostname,port,delimiter);


	        // 3 transformation 
	        DataStream<WordWithCount> windowCounts = text
	            .flatMap(new FlatMapFunction<String, WordWithCount>() {
	            @Override
	            public void flatMap(String value, Collector<WordWithCount> out)
	                    throws Exception {
	                String[] splits = value.split("\\s");  
	                for (String word : splits){
	                    out.collect(new WordWithCount(word , 1L));    
	                }

	            }
	        })
	        .keyBy("word")
	        .timeWindow(Time.seconds(5), Time.seconds(1))
	        .reduce(new ReduceFunction<WordWithCount>() {
	            public WordWithCount reduce(WordWithCount a, WordWithCount b) {
	                return new WordWithCount(a.word, a.count + b.count);
	            }
	        });
	        // timeWindow(Time size, Time slide)

	        windowCounts.print().setParallelism(1);

	        env.execute("Socket window count run");
	    }

	    public static class WordWithCount{
	        public String word ;
	        public long count ;

	        public  WordWithCount(){}

	        public WordWithCount(String word, long count){
	            this.word = word ;
	            this.count = count ;
	        }

	        @Override
	        public String toString() {
	            return "WordWithCount{" +
	                    "word='" + word + '\'' +
	                    ", count=" + count +
	                    '}';
	        }
	    }
}
